package com.shujia.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo22Broadcast {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("broadcast")
      .setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val students: RDD[String] = sc.textFile("data/students.txt")

    val score: RDD[String] = sc.textFile("data/score.txt")

    /**
      * 广播变量的应用
      *
      * 实现map join
      * 将小表加载内存中，在map端进行关联
      *
      */

    //collect : 将rdd的数据拉去到Driver端的内存中
    val listStudents: Array[String] = students.collect

    val studentMap = listStudents.map(stu => {

      val student: Array[String] = stu.split(",")

      (student(0),stu)

    }).toMap

    //将student小表广播
    val broadCastMap: Broadcast[Map[String, String]] = sc.broadcast(studentMap)

    val stuCoInfo: RDD[String] = score.map(sco => {

      val id: String = sco.split(",")(0)

      val stu: Map[String, String] = broadCastMap.value

      val stuInfo: String = stu.getOrElse(id,"未找到")

      stuInfo + "\t" + sco

    })

    stuCoInfo.foreach(println)

  }

}
